package com.lvlian;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.utils.JdbcUtils;
import com.utils.OSUtils;
import com.utils.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.HashMap;
import java.util.Map;

public class LvlianSinkVehicle extends RichSinkFunction<String> {
    private static final Logger LOG = LoggerFactory.getLogger(LvlianSinkVehicle.class);
    private Connection connection ;
    private PreparedStatement preparedStatementInsert;
    private PreparedStatement preparedStatementDelete;
    private String lastValue = null;
    private Map<Integer, KV> map = new HashMap<>();

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        // 创建 PreparedStatement 对象
        String sqlInsert = "replace INTO lv_vehicle_info ( " +
                "id,plate_no,plate_color,vehicle_type,identify_code," +
                "vehicle_owner,use_nature,vehicle_brand,engine_no," +
                "registration_date,approved_date,total_weight,empty_weight,load_capacity," +
                "expiration_date,agency,create_user,create_date,org_code" +
                 ") VALUES (" +
                "?, ?, ?, ?, ?," +
                " ?, ?, ?, ?," +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?" +
                ")";

        connection =    JdbcUtils.getInstance().getConnection();
        LOG.info(sqlInsert);
        String sqlDelete = "DELETE FROM lv_vehicle_info WHERE id = ?";
        LOG.info(sqlInsert);
        preparedStatementInsert = connection.prepareStatement(sqlInsert);
        preparedStatementDelete = connection.prepareStatement(sqlDelete);


        map.put(1, new KV("ROW_ID", "id"));
        map.put(2, new KV("TRUCK_NUM", "plate_no"));
        map.put(3, new KV("TRUCK_COLOUR_NUM", "plate_color"));
        map.put(4, new KV("TRUCK_TYPE", "vehicle_type"));
        map.put(5, new KV("TRUCK_VIN", "identify_code"));

        map.put(6, new KV("OWNER", "vehicle_owner"));
        map.put(7, new KV("USE", "use_nature"));
        map.put(8, new KV("TRUCK_BRAND_KIND", "vehicle_brand"));
        map.put(9, new KV("ENGINE_NUM", "engine_no"));
        map.put(10, new KV("REGISTRATION_DATE", "registration_date"));
        map.put(11, new KV("APPROVED_DATE", "approved_date"));
        map.put(12, new KV("TOTAL_WEIGHT", "total_weight"));
        map.put(13, new KV("EMPTY_WEIGHT", "empty_weight"));
        map.put(14, new KV("LOAD_CAPACITY", "load_capacity"));
        map.put(15, new KV("EXPIRATION_DATE", "expiration_date"));
        map.put(16, new KV("AGENCY", "agency"));
        map.put(17, new KV("RECORDER", "create_user"));
        map.put(18, new KV("RECORD_DATE", "create_date"));
        map.put(19, new KV("ORG_CODE", "org_code"));

    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        // 将 JSON 字符串解析为 JSONObject
        JSONObject jsonObject = JSON.parseObject(value);

        // 获取操作类型
        String operation = jsonObject.getString("op");

        // 判断操作类型
        if ("d".equals(operation)) {
            // 删除操作
            JSONObject beforeObject = jsonObject.getJSONObject("before");
            String id = beforeObject.getString("ROW_ID");
            preparedStatementDelete.setString(1, id);
            preparedStatementDelete.executeUpdate();
        } else {
            // 插入或更新操作
            JSONObject afterObject = jsonObject.getJSONObject("after");
            // 如果上一次处理的值为空，或者当前值与上一次处理的值不相同，则执行 sink 操作
            if (lastValue == null || !lastValue.equals(value)) {
                // 遍历原始表字段
                for (Map.Entry<Integer, KV> entry : map.entrySet()) {
                    String v = afterObject.getString(entry.getValue().getK());
                    if ("RECYCLING_IN_DATE".equals(entry.getValue().getK())
                            || "REGISTRATION_DATE".equals(entry.getValue().getK())
                            || "APPROVED_DATE".equals(entry.getValue().getK())
                            || "EXPIRATION_DATE".equals(entry.getValue().getK())
                            || "RECORD_DATE".equals(entry.getValue().getK())) {
                        v = StringUtils.restoreDate(v);
                    }
                    LOG.info(entry.getKey() + ": " + entry.getValue().getK() + " -> " + entry.getValue().getV() + " -> " + v);
                    preparedStatementInsert.setString(entry.getKey(), v);
                }

                preparedStatementInsert.executeUpdate();

                // 更新上一次处理的值为当前值
                lastValue = value;
            }
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭连接和 PreparedStatement
        if (preparedStatementInsert != null) {
            preparedStatementInsert.close();
        }
        if (preparedStatementDelete != null) {
            preparedStatementDelete.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}
